package io.gitee.tziye;

import io.gitee.tziye.common.RenaiUtil;
import io.gitee.tziye.core.bloom.BloomOperator;
import io.gitee.tziye.core.fs.FsOperator;
import io.gitee.tziye.core.id.IdOperator;
import io.gitee.tziye.core.instance.InstanceSpec;
import io.gitee.tziye.core.instance.RenaiInstance;
import io.gitee.tziye.core.kv.ListOperator;
import io.gitee.tziye.core.kv.MapOperator;
import io.gitee.tziye.core.kv.PairOperator;
import io.gitee.tziye.core.kv.PriorityQOperator;
import io.gitee.tziye.core.kv.model.RenaiList;
import io.gitee.tziye.core.kv.model.RenaiMap;
import io.gitee.tziye.core.kv.model.RenaiPair;
import io.gitee.tziye.core.kv.model.RenaiPriorityQ;
import io.gitee.tziye.core.lock.LockOperator;
import io.gitee.tziye.core.lock.RenaiLock;
import io.gitee.tziye.core.msg.MsgOperator;
import io.gitee.tziye.core.msg.RenaiTopic;
import io.gitee.tziye.core.search.RenaiSearch;
import io.gitee.tziye.core.search.SearchOperator;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Supplier;

import static io.gitee.tziye.common.Constants.DELIMITER;

public abstract class RenaiClient {

    public final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(
            Runtime.getRuntime().availableProcessors() / 2);

    /**
     * Instance
     */
    public final RenaiInstance instance;

    public RenaiClient(InstanceSpec instanceSpec) {
        instance = new RenaiInstance(UUID.randomUUID().toString(), RenaiUtil.localIp(), instanceSpec);
        register();
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }

    @Setter
    @Getter
    protected volatile boolean enableRegister = true;

    protected abstract void register();

    protected abstract void shutdown();

    public abstract RenaiInstance realStatus();

    public abstract List<RenaiInstance> instances();

    public abstract List<RenaiInstance> lookup(String group, String service);

    /**
     * Operator
     */

    protected PairOperator pairOperator;
    protected ListOperator listOperator;
    protected PriorityQOperator priorityQOperator;
    protected MapOperator mapOperator;
    protected MsgOperator msgOperator;
    protected LockOperator lockOperator;
    protected SearchOperator searchOperator;
    protected FsOperator fsOperator;
    protected IdOperator idOperator;
    protected BloomOperator bloomOperator;

    public PairOperator pair() {
        return pairOperator;
    }

    public ListOperator list() {
        return listOperator;
    }

    public PriorityQOperator priorityQ() {
        return priorityQOperator;
    }

    public MapOperator map() {
        return mapOperator;
    }

    public MsgOperator msg() {
        return msgOperator;
    }

    public LockOperator lock() {
        return lockOperator;
    }

    public SearchOperator search() {
        return searchOperator;
    }

    public FsOperator fs() {
        return fsOperator;
    }

    public IdOperator id() {
        return idOperator;
    }

    public BloomOperator bloom() {
        return bloomOperator;
    }

    /**
     * KV
     */

    public <T> RenaiPair<T> pair(String key, T value, Class<T> clazz) {
        return new RenaiPair<>(this, key, value, clazz, null, null);
    }

    public <T> RenaiPair<T> pair(String key, T value, Class<T> clazz, int expireSeconds) {
        return new RenaiPair<>(this, key, value, clazz, expireSeconds, null);
    }

    public <T> RenaiPair<T> pair(String key, T value, Class<T> clazz, Integer expireSeconds, int localCacheSeconds) {
        return new RenaiPair<>(this, key, value, clazz, expireSeconds, localCacheSeconds);
    }

    public <T> RenaiList<T> list(String key, Class<T> clazz) {
        return new RenaiList<>(this, key, clazz, null);
    }

    public <T> RenaiList<T> list(String key, Class<T> clazz, int expireSeconds) {
        return new RenaiList<>(this, key, clazz, expireSeconds);
    }

    public <T> RenaiPriorityQ<T> priorityQ(String key, Class<T> clazz) {
        return new RenaiPriorityQ<>(this, key, clazz, null);
    }

    public <T> RenaiPriorityQ<T> priorityQ(String key, Class<T> clazz, int expireSeconds) {
        return new RenaiPriorityQ<>(this, key, clazz, expireSeconds);
    }

    public <T> RenaiMap<T> map(String key, Class<T> clazz) {
        return new RenaiMap<>(this, key, clazz, null);
    }

    public <T> RenaiMap<T> map(String key, Class<T> clazz, int expireSeconds) {
        return new RenaiMap<>(this, key, clazz, expireSeconds);
    }

    /**
     * Message
     */

    public <T> RenaiTopic<T> topic(String topic, Class<T> clazz) {
        String defaultGroup = instance.getG() + DELIMITER + instance.getService();
        return new RenaiTopic<>(this, topic, defaultGroup, clazz);
    }

    public <T> RenaiTopic<T> topic(String topic, String group, Class<T> clazz) {
        return new RenaiTopic<>(this, topic, group, clazz);
    }

    /**
     * Lock
     */

    public RenaiLock lock(String key, int expireSeconds) {
        RenaiLock lock = new RenaiLock(this, key, expireSeconds, null);
        lock.lock();
        return lock;
    }

    public RenaiLock lock(String key, int expireSeconds, int timeout) {
        RenaiLock lock = new RenaiLock(this, key, expireSeconds, timeout);
        lock.lock();
        return lock;
    }

    /**
     * Executor
     */

    public <T> T execute(String key, int expireSeconds, Supplier<T> supplier) {
        return commonExecute(key, expireSeconds, null, supplier, true);
    }

    public <T> T executeNoRelease(String key, int expireSeconds, Supplier<T> supplier) {
        return commonExecute(key, expireSeconds, null, supplier, false);
    }

    public <T> T execute(String key, int expireSeconds, int timeout, Supplier<T> supplier) {
        return commonExecute(key, expireSeconds, timeout, supplier, true);
    }

    public <T> T executeNoRelease(String key, int expireSeconds, int timeout, Supplier<T> supplier) {
        return commonExecute(key, expireSeconds, timeout, supplier, false);
    }

    public <T> T commonExecute(String key, int expireSeconds, Integer timeout, Supplier<T> supplier, boolean release) {
        assert StringUtils.isNotBlank(key) && expireSeconds > 0 && supplier != null && (timeout == null || timeout > 0);
        RenaiLock lock;
        if (timeout == null) {
            lock = lock(key, expireSeconds);
        } else {
            lock = lock(key, expireSeconds, timeout);
        }
        if (!lock.isLocked()) {
            return null;
        }
        try {
            return supplier.get();
        } finally {
            if (release) {
                lock.unlock();
            }
        }
    }

    /**
     * Search
     */
    public <T> RenaiSearch<T> search(String module, Class<T> clazz) {
        return new RenaiSearch<>(this, module, clazz);
    }

}
